package com.mqtt.config;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.util.UUID;

/**
 * MQTT配置类
 */
@Slf4j
@Data
@Configuration
public class MqttConfiguration {
    
    @Value("${mqtt.broker.url:tcp://localhost:1883}")
    private String brokerUrl;
    
    @Value("${mqtt.client.id:mqtt-service-}")
    private String clientId;
    
    @Value("${mqtt.username:admin}")
    private String username;
    
    @Value("${mqtt.password:admin123}")
    private String password;
    
    @Value("${mqtt.topics:device/+/data,device/+/status,device/+/command}")
    private String[] topics;
    
    @Value("${mqtt.qos:1}")
    private int qos;
    
    @Value("${mqtt.connection.timeout:30}")
    private int connectionTimeout;
    
    @Value("${mqtt.keepalive.interval:60}")
    private int keepAliveInterval;
    
    @Value("${mqtt.automatic.reconnect:true}")
    private boolean automaticReconnect;
    
    @Value("${mqtt.clean.session:true}")
    private boolean cleanSession;
    
    /**
     * MQTT客户端工厂
     */
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        
        options.setServerURIs(new String[]{brokerUrl});
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(connectionTimeout);
        options.setKeepAliveInterval(keepAliveInterval);
        options.setAutomaticReconnect(automaticReconnect);
        options.setCleanSession(cleanSession);
        
        // 设置遗嘱消息
        String willTopic = "service/status/" + clientId;
        String willMessage = "{\"status\":\"offline\",\"timestamp\":" + System.currentTimeMillis() + "}";
        options.setWill(willTopic, willMessage.getBytes(), 1, true);
        
        factory.setConnectionOptions(options);
        log.info("MQTT Client Factory initialized with broker: {}", brokerUrl);
        
        return factory;
    }
    
    /**
     * 入站通道 - 接收MQTT消息
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }
    
    /**
     * 出站通道 - 发送MQTT消息
     */
    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }
    
    /**
     * MQTT消息接收适配器
     */
    @Bean
    public MessageProducer mqttInbound() {
        String inboundClientId = clientId + "-inbound-" + UUID.randomUUID().toString().substring(0, 8);
        MqttPahoMessageDrivenChannelAdapter adapter = 
            new MqttPahoMessageDrivenChannelAdapter(inboundClientId, mqttClientFactory(), topics);
        
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(qos);
        adapter.setOutputChannel(mqttInputChannel());
        
        log.info("MQTT Inbound adapter initialized with topics: {}", String.join(", ", topics));
        
        return adapter;
    }
    
    /**
     * MQTT消息发送处理器
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        String outboundClientId = clientId + "-outbound-" + UUID.randomUUID().toString().substring(0, 8);
        MqttPahoMessageHandler messageHandler = 
            new MqttPahoMessageHandler(outboundClientId, mqttClientFactory());
        
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(qos);
        messageHandler.setDefaultRetained(false);
        
        log.info("MQTT Outbound handler initialized");
        
        return messageHandler;
    }
    
    /**
     * 添加动态主题订阅
     */
    public void addTopicSubscription(String topic) {
        if (mqttInbound() instanceof MqttPahoMessageDrivenChannelAdapter) {
            MqttPahoMessageDrivenChannelAdapter adapter = 
                (MqttPahoMessageDrivenChannelAdapter) mqttInbound();
            adapter.addTopic(topic, qos);
            log.info("Added subscription to topic: {}", topic);
        }
    }
    
    /**
     * 移除主题订阅
     */
    public void removeTopicSubscription(String topic) {
        if (mqttInbound() instanceof MqttPahoMessageDrivenChannelAdapter) {
            MqttPahoMessageDrivenChannelAdapter adapter = 
                (MqttPahoMessageDrivenChannelAdapter) mqttInbound();
            adapter.removeTopic(topic);
            log.info("Removed subscription from topic: {}", topic);
        }
    }
}